package com.lmf.cloud.topic;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.DeleteTopicsOptions;
import org.apache.kafka.clients.admin.DeleteTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.TopicConfig;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @Description kafka主题操作实例
 * @Author LiuMingFu
 * @Date 2024/8/10 11:49
 */
//@Slf4j
public class KafkaTopicMain {
    public static void main(String[] args) {
        // Kafka 集群配置
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        //创建kafka客户端
        AdminClient adminClient = AdminClient.create(properties);

        // 主题分区参数
        int partitions = 3;
        // 主题副本数，一般为2，且不能超过集群节点数，否则创建主题失败
        short replicas = 1;
        //主题名称
        String topicName = "depart-2";

        //调用创建主题方法
        createTopic(adminClient, topicName, partitions, replicas);

        //调用删除主题方法
//        deleteTopic(adminClient, topicName);
    }


    /**
     * 创建主题
     *
     * @param adminClient kafka客户端
     * @param topicName   主题名称
     * @param partitions  分区数
     * @param replicas    副本数
     * @return
     * @author LiuMingFu
     * @date 2024/8/10
     */
    public static void createTopic(AdminClient adminClient, String topicName, int partitions, short replicas) {
        //一个主题实例
        NewTopic newTopic = new NewTopic(topicName, partitions, replicas);

        //添加一些额外的配置
        Map<String, String> configs = new HashMap<>();
        //清理策略：超过7天的数据自动删除
        configs.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE);
        newTopic.configs(configs);

        //封装主题集合
        List<NewTopic> topics = Arrays.asList(newTopic);

        //执行创建主题集合
        adminClient.createTopics(topics);

        //关闭kafka客户端
        adminClient.close();

    }


    /**
     * 删除主题
     *
     * @param adminClient   kafka连接客户端
     * @param topicToDelete 需要删除的主题名称
     * @return
     * @author LiuMingFu
     * @date 2024/8/10
     */
    public static void deleteTopic(AdminClient adminClient, String topicToDelete) {
        //配置删除主题的额外参数
        DeleteTopicsOptions options = new DeleteTopicsOptions();
        options.timeoutMs(10000);

        //执行异步删除主题
        DeleteTopicsResult result = adminClient.deleteTopics(Arrays.asList(topicToDelete), options);

        try {
            //等待完全删除成功
            result.all().get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

        //关闭kafka客户端连接
        adminClient.close();
    }
}
